package com.dark.trident;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import storm.trident.Stream;
import storm.trident.TridentTopology;

/**
 * Created by tengxue on 16-7-3.
 */
public class OutbreakDetectionTopology {

    public static StormTopology buildTopology(){
        TridentTopology topology=new TridentTopology();
        DiagnosisEventSpout spout=new DiagnosisEventSpout();
        Stream inputStream=topology.newStream("event",spout);
        inputStream
                //Filter for critical events.
                .each(new Fields("event"),new DiseaseFilter())
                //Locate the closest city
                .each(new Fields("event"), new CityAssignment(), new Fields("city"))
                //Derive the hour segment
                .each(new Fields("event", "city"), new HourAssignment(), new Fields("hour","cityDiseaseHour"))
                //Group occurrences in same city and hour
                .groupBy(new Fields("cityDiseaseHour"))
                //Count occurrences and persist the results
                .persistentAggregate(new OutbreakTrendFactory(), new Count(), new Fields("count"))
                .newValuesStream()
                //Detect an outbreak
                .each(new Fields("cityDiseaseHour","count"),new OutbreakDetector(),new Fields("alert"))
                //Dispatch the alert
                .each(new Fields("alert"), new DispatchAlter(), new Fields());

        return topology.build();
    }

    public static void main(String[] args)throws Exception{
        Config config=new Config();

        //建议加上这行，使得每个bolt/spout的并发度都为1
        config.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 1);
        LocalCluster cluster = new LocalCluster();

        //提交拓扑
        cluster.submitTopology("cdc", config, buildTopology());

        //本地调试的时候必须添加这段代码.否则会出现zookeeper连接拒绝的异常
        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        //结束拓扑
        cluster.killTopology("cdc");

        cluster.shutdown();
    }
}
